<?php

namespace App\Http\Common\traits;

use App\Http\Common\Contexts\SwooleDemonContext;
use App\Logging\Processor\UidProcessor;
use Illuminate\Support\Facades\Log;
use Swoole\Coroutine;
use Swoole\Coroutine\Channel;
use Swoole\Process;
use Swoole\Runtime;
use Throwable;
use function Swoole\Coroutine\run;

trait BasicTaskScheduler
{

    /** @var Channel */
    private $channel;

    /**
     * 进程配置
     * [
     *   'task1' => [
     *     'count' => 1,
     *     'handler' => [$this, 'task1'],
     *   ],
     *   'task2' => [
     *     'count' => 5,
     *     'handler' => [$this, 'task2'],
     *   ],
     * ]
     * @var array
     */
    private $config;

    /** @var bool */
    private $finish = false;

    /**
     * Execute the console command.
     *
     * @return int
     */
    public function handle()
    {
        // 此行代码后，文件操作，sleep，Mysqli，PDO，streams等都变成异步IO
        Runtime::enableCoroutine();
        // run 创建一个协程容器
        run(function () {
            $this->registerSignal();
            // 协程间的消息队列，多个协程通过 push 和 pop 操作生产消息和消费消息，用来协程之间的通讯。
            $this->channel = new Channel(1024);
            $this->init();
        });
        return 0;
    }

    private function registerSignal(): void
    {
        Process::signal(SIGINT, [$this, 'signal']);
        Process::signal(SIGTERM, [$this, 'signal']);
    }

    public function signal(int $sig)
    {
        switch ($sig) {
            case SIGINT:
                $this->end(PHP_INT_MAX);
                break;

            case SIGTERM:
                $this->end(10);
                break;

            default:
                Log::info("无法处理的信号: {$sig}");
                break;
        }
    }

    private function end(int $wait): void
    {
        echo "finish";
        $this->finish = true;
        for ($i = 0; $i < $wait; $i++) {
            echo ".";
            if (count(iterator_to_array(Coroutine::listCoroutines())) <= 1) {
                return;
            }
            sleep(1);
        }
        echo PHP_EOL;

        Process::kill(posix_getpid(), SIGKILL);
    }

    private function init()
    {
        // 创建协程 Coroutine::create 或 go 方法创建协程
        // 主协程
        Coroutine::create(function () {
            $_workers_idx_cid = SwooleDemonContext::getWorkersIdxCid();
            foreach ($this->config as $k => $config) {
                for ($i = 0; $i < $config['count']; $i++) {
                    $cid = $_workers_idx_cid[$k][$i] ?? null;
                    if (is_null($cid) || !Coroutine::exists($cid)) {
                        $this->startTask($i, $k);
                    }
                }
            }
        });

        while (1) {
            if ($this->finish) {
                break;
            }

            // pop超时时间必须大于最大的协程执行时间，否则channel会抛出致命异常：all coroutines (count: xxx) are asleep - deadlock!
            $data = $this->channel->pop(-1);
            if ($data) {
                Log::info('task done: ' . json_encode($data, JSON_UNESCAPED_UNICODE) . '.');
                Log::debug('task restart.');
                $this->startTask($data['idx'], $data['key']);
                Log::info('task restart success.');
            } else if ($this->channel->errCode === SWOOLE_CHANNEL_TIMEOUT) {
                break;
            }
        }

        if (!$this->finish) {
            $this->init();
        }
    }

    private function startTask(int $idx, string $key)
    {
        Coroutine::create(function () use ($idx, $key) {
            SwooleDemonContext::setWorkersIdxCid($key, $idx, Coroutine::getCid());
            // 协程回收资源的时候调用defer
            Coroutine::defer(function () use ($idx, $key) {
                Log::getLogger()->close();
                $this->channel->push([
                    'idx' => $idx,
                    'key' => $key,
                ]);
            });

            $args = $this->config[$key]['args'];
            $args[] = $idx;

            while (1) {
                if ($this->finish) {
                    break;
                }

                try {
                    UidProcessor::reset();
                    call_user_func_array($this->config[$key]['handler'], $args);

                } catch (Throwable $e) {
                    Log::error("call handler[{$key}][{$idx}] fail: {$e->getMessage()}", [
                        'exception' => $e
                    ]);
                }
            }
        });
    }

}
